package cn.com.zetatech.loader.common.service.impl;

import cn.com.zetatech.loader.common.service.KafkaService;
import org.apache.kafka.clients.consumer.*;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.header.Header;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.ConsumerFactory;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.kafka.support.TopicPartitionOffset;
import org.springframework.stereotype.Service;
import org.springframework.util.concurrent.ListenableFuture;

import java.util.*;

/**
 * @author zhengbs
 */
@Service
public class KafkaServiceImpl implements KafkaService {

    @Autowired
    private KafkaTemplate kafkaTemplate;

    @Autowired
    private ConsumerFactory consumerFactory;

    @Override
    public RecordMetadata sendMessage(String topic, String message, List<Header> headers) throws Exception {
        ProducerRecord<String, String> producerRecord = new ProducerRecord<>(topic, null, null, null, message, headers);
        ListenableFuture<SendResult<String, String>> sendResult = kafkaTemplate.send(producerRecord);
        SendResult<String, String> result = sendResult.get();
        return result.getRecordMetadata();
    }

    @Override
    public RecordMetadata sendMessage(String topic, String key, String message, List<Header> headers) throws Exception {
        ProducerRecord<String, String> producerRecord = new ProducerRecord<>(topic, null, null, key, message, headers);
        ListenableFuture<SendResult<String, String>> sendResult = kafkaTemplate.send(producerRecord);
        SendResult<String, String> result = sendResult.get();
        return result.getRecordMetadata();
    }

    @Override
    public void commit(String groupId, Map<TopicPartition, OffsetAndMetadata> offsets) {

        Consumer consumer = consumerFactory.createConsumer(groupId, "");
        consumer.commitAsync(offsets, new OffsetCommitCallback() {
            @Override
            public void onComplete(Map<TopicPartition, OffsetAndMetadata> offsets, Exception exception) {
                if (exception != null) {
                    System.err.println("Commit failed for" + offsets);
                }
            }
        });
    }

}
